// @Author EthanScriptOn
// @Desc
package connect

import (
	"fmt"
	"gitee.com/fatzeng/srf_switch_basic_components/util"
	"gitee.com/fatzeng/srf_switch_expansion_package/detect"
	"github.com/apache/rocketmq-client-go/v2"
)

var RocketmqProducer *util.OnceInitializer

var RocketmqResourceChecker = func() (isInvalid bool) {
	producer, err := generateRocketmqProducer()
	if err != nil {
		return true
	}
	err = producer.Start()
	if err != nil {
		_ = producer.Shutdown()
		return true
	}
	return false
}

var RocketmqResourceRecover = func() (err error) {
	RocketmqProducer.Reset()
	_, err = RocketmqProducer.Get()
	return err
}

var RocketmqProducerMonitor = detect.GenerateResourceMonitor("rocketmq_producer", true, RocketmqResourceChecker, RocketmqResourceRecover, detect.MonitorFunctionInstance)

func GetRocketmqProducer() (producer rocketmq.Producer, err error) {
	defer detect.BackendHealthChecker.AddResource(RocketmqProducerMonitor)
	return generateRocketmqProducer()
}

func generateRocketmqProducer() (producer rocketmq.Producer, err error) {
	if RocketmqProducer == nil {
		return nil, fmt.Errorf("the RocketmqProducer initialization required")
	}
	instanceInter, err := RocketmqProducer.Get()
	if err != nil {
		return nil, err
	}
	instance, ok := instanceInter.(rocketmq.Producer)
	if !ok {
		return nil, fmt.Errorf("the instance is not of type Producer")
	}
	return instance, nil
}

var RocketmqConsumer *util.OnceInitializer

var RocketmqConsumerChecker = func() (isInvalid bool) {
	consumer, err := generateRocketmqConsumer()
	if err != nil {
		return true
	}
	err = consumer.Start()
	if err != nil {
		_ = consumer.Shutdown()
		return true
	}
	return false
}

var RocketmqConsumerRecover = func() (err error) {
	RocketmqConsumer.Reset()
	_, err = RocketmqConsumer.Get()
	return err
}

var RocketmqConsumerMonitor = detect.GenerateResourceMonitor("rocket_mq_consumer", true, RocketmqConsumerChecker, RocketmqConsumerRecover, detect.MonitorFunctionInstance)

func GetRocketmqConsumer() (producer rocketmq.PushConsumer, err error) {
	defer detect.BackendHealthChecker.AddResource(RocketmqConsumerMonitor)
	return generateRocketmqConsumer()
}

func generateRocketmqConsumer() (consumer rocketmq.PushConsumer, err error) {
	if RocketmqConsumer == nil {
		return nil, fmt.Errorf("the RocketmqConsumer initialization required")
	}
	instanceInter, err := RocketmqConsumer.Get()
	if err != nil {
		return nil, err
	}
	instance, ok := instanceInter.(rocketmq.PushConsumer)
	if !ok {
		return nil, fmt.Errorf("the instance is not of type PushConsumer")
	}
	return instance, nil
}
